package com.imooc.disruptor.heigh.multi;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;

/**
 * @author yqc 2018/10/11
 */
@Slf4j
public class Main {

    public static void main(String[] args) throws Exception {

        //1 创建RingBuffer
        RingBuffer<Order> ringBuffer = RingBuffer.create(ProducerType.MULTI,
                Order::new, 1024 * 1024, new YieldingWaitStrategy());

        //2 通过ringBuffer创建一个屏障
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        //3 创建多个消费者数组
        Consumer[] consumers = new Consumer[10];
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = new Consumer("C_" + i);
        }

        //4 构建多消费者工作池
        WorkerPool<Order> workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier,
                new EventExceptionHandler(), consumers);

        //5 设置多个消费者的sequence序号，用于单独统计消费进度
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        //6 启动workPool
        int nThreads = Runtime.getRuntime().availableProcessors();
        log.info("nThreads:{}", nThreads);
        workerPool.start(Executors.newFixedThreadPool(nThreads));

        CountDownLatch latch = new CountDownLatch(1);

        for (int i = 0; i < 100; i++) {
            Producer producer = new Producer(ringBuffer);
            new Thread(() -> {
                try {
                    latch.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                for (int j = 0; j < 100; j++) {
                    producer.setData(UUID.randomUUID().toString());
                }
            }).start();
        }

        Thread.sleep(2000);
        log.info("----------线程创建完毕，开始生产数据----------");
        latch.countDown();

        Thread.sleep(5000);

        log.info("任务总数：{}", Consumer.getCount());
    }

    static class EventExceptionHandler implements ExceptionHandler<Order> {

        @Override
        public void handleEventException(Throwable ex, long sequence, Order event) {

        }

        @Override
        public void handleOnStartException(Throwable ex) {

        }

        @Override
        public void handleOnShutdownException(Throwable ex) {

        }
    }

}
